//package com.central.service
//
//import java.sql.Timestamp
//import java.text.SimpleDateFormat
//import java.util
//import java.util.{Date, Properties, UUID}
//
//import com.central.bean
//import com.central.bean.{ClueRdd, ClueResultWithDetail, WifiEntity}
//import com.central.controller.{ActivityTime, InitComplexClueConditionEntity}
//import com.central.utils.{DateFormat, LoadESData, ParseTime}
//import com.central.utils.Relationship._
//import org.apache.spark.rdd.RDD
//import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
//
//import scala.collection.Map
//import scala.reflect.ClassTag
//import scala.reflect.runtime.universe.TypeTag
////import scala.reflect.ClassTag
//import scala.util.{Failure, Success, Try}
//
//class ClueService {
//  val url = Some("jdbc:mysql://10.68.2.237:3306/ga?characterEncoding=utf-8&serverTimezone=UTC") //getPropertyAsOption("lx.result.url")
//  require(url.isDefined)
//  val user = Some("root") //getPropertyAsOption("lx.result.username")
//  require(user.isDefined)
//  val password = Some("root") //getPropertyAsOption("lx.result.password")
//  require(password.isDefined)
//
//  val hybaseUrl = Some("http://10.68.0.230:5555") //getPropertyAsOption("hybase.url")
//  require(hybaseUrl.isDefined)
//  val hybaseUser = Some("test") //getPropertyAsOption("hybase.username")
//  require(hybaseUser.isDefined)
//  val hybasePassword = Some("trsadmin123") //getPropertyAsOption("hybase.password")
//  require(hybasePassword.isDefined)
//
//  /**
//    * 碰撞
//    *
//    * @param spark
//    * @param jobId
//    * @param conditions
//    * @param threshold
//    */
//  def clueold(spark: SparkSession, jobId: String, conditions: Array[ClueConditionEntity], threshold: Int): Unit = {
//    require(conditions != null)
//    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
//    var rddWfifEntity: RDD[WifiEntity] = null
//    conditions.map(s => {
//      if (rddWfifEntity == null) {
//        rddWfifEntity = LoadESData.getOneDeviceData(spark, "lzwifi/wifi", s.deviceNumber, sdf.format(new Date(s.beginTime)), sdf.format(new Date(s.endTime))).map(e => {
//          if (e._2.get("mac").get != None) //mac数据的封装
//            WifiEntity(
//              e._1, e._2.get("mac").get + "", e._2.get("mac") + "", "MAC",
//              Timestamp.valueOf(e._2.get("captime") + ""), Timestamp.valueOf(e._2.get("captime") + ""),
//              //              Timestamp.valueOf(e._2.get("ir_time") + ""),
//              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//              e._2.get("device_code").get + "",
//              e._2.get("device_name").get + "",
//              e._2.get("address").get + "",
//              e._2.get("location") + "",
//              e._2.get("captime").get + "",
//              java.lang.Double.parseDouble(e._2.get("latitude") + ""),
//              java.lang.Double.parseDouble(e._2.get("longtitude") + ""),
//
//              e._2.get("phone").get + "", "",
//              e._2.get("picpath").get + ""
//            )
//          else
//            WifiEntity(
//              e._1, e._2.get("imsi").get + "", e._2.get("imsi") + "", "IMSI",
//              Timestamp.valueOf(e._2.get("captime") + ""), Timestamp.valueOf(e._2.get("captime") + ""),
//              //              Timestamp.valueOf(e._2.get("ir_time") + ""),
//              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//              e._2.get("device_code").get + "",
//              e._2.get("device_name").get + "",
//              e._2.get("address").get + "",
//              e._2.get("location") + "",
//              e._2.get("captime").get + "",
//              java.lang.Double.parseDouble(e._2.get("latitude") + ""),
//              java.lang.Double.parseDouble(e._2.get("longtitude") + ""),
//              e._2.get("phone").get + "", "",
//              e._2.get("picpath").get + ""
//            )
//
//        })
//      } else {
//        rddWfifEntity = rddWfifEntity ++ LoadESData.getOneDeviceData(spark, "lzwifi/wifi", s.deviceNumber, sdf.format(new Date(s.beginTime)), sdf.format(new Date(s.endTime))).map(e => {
//          if (e._2.get("mac").get != None) //mac数据的封装
//            WifiEntity(
//              e._1, e._2.get("mac").get + "", e._2.get("mac") + "", "MAC",
//              Timestamp.valueOf(e._2.get("captime") + ""), Timestamp.valueOf(e._2.get("captime") + ""),
//              //              Timestamp.valueOf(e._2.get("ir_time") + ""),
//              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//              e._2.get("device_code").get + "",
//              e._2.get("device_name").get + "",
//              e._2.get("address").get + "",
//              e._2.get("location") + "",
//              e._2.get("captime").get + "",
//
//              java.lang.Double.parseDouble(e._2.get("latitude") + ""),
//              java.lang.Double.parseDouble(e._2.get("longtitude") + ""),
//
//              e._2.get("phone").get + "", "",
//              e._2.get("picpath").get + ""
//            )
//          else
//            WifiEntity(
//              e._1, e._2.get("imsi").get + "", e._2.get("imsi") + "", "IMSI",
//              Timestamp.valueOf(e._2.get("captime") + ""), Timestamp.valueOf(e._2.get("captime") + ""),
//              //              Timestamp.valueOf(e._2.get("ir_time") + ""),
//              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//              e._2.get("device_code").get + "",
//              e._2.get("device_name").get + "",
//              e._2.get("address").get + "",
//              e._2.get("location") + "",
//              e._2.get("captime").get + "",
//
//              java.lang.Double.parseDouble(e._2.get("latitude") + ""),
//              java.lang.Double.parseDouble(e._2.get("longtitude") + ""),
//
//              e._2.get("phone").get + "", "",
//              e._2.get("picpath").get + ""
//            )
//
//        })
//      }
//    })
//    //进行碰撞
//    val clueResultRdd: RDD[ClueResultWithDetail] = new ClueRdd(rddWfifEntity).clue(spark, jobId, threshold)
//    val properties = new Properties()
//    properties.put("user", "root")
//    properties.put("password", "root")
//    spark.createDataFrame(clueResultRdd).write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.5.180:3306/ga?characterEncoding=utf-8&serverTimezone=UTC", "T_CLUE_RESULT", properties)
//  }
//
//
//  //  def clue(spark: SparkSession, clueContext: InitComplexClueConditionEntity): Unit = {
//  //    require(clueContext != null)
//  //    val conditions: List[(List[ClueConditionEntity], Int)] =
//  //      clueContext
//  //        .conditions
//  //        .map(listConditions => {
//  //          val where = listConditions
//  //            .map(condition => ClueConditionEntity(condition.deviceNumber, condition.begintime, condition.endtime, if (condition.didAppear == 0) false else true))
//  //          val threshold = listConditions.head.threshold
//  //          (where, threshold)
//  //        })
//  //    val threshold = conditions.head._2
//  //    val conditionsRange = clueContext.conditions.size
//  //    val relation = clueContext.relationship //or and not
//  //
//  //    //对每个碰撞节点进行处理并根据relationship关系对碰撞结果进行合并
//  //    val clueOriginalResultRdd: RDD[ClueResultWithDetail] = conditions
//  //
//  //      .foldLeft(spark.sparkContext.emptyRDD[ClueResultWithDetail]) {
//  //        case (prevRdd, oneNodeConditions) => {
//  //
//  //          val currNodeRdd = justClue(spark, clueContext.jobId, oneNodeConditions._1.toArray[ClueConditionEntity], oneNodeConditions._2)
//  //          if (prevRdd.isEmpty() && !currNodeRdd.isEmpty())
//  //            currNodeRdd
//  //          else if (!currNodeRdd.isEmpty()) {
//  //            val rdd: RDD[ClueResultWithDetail] = currNodeRdd
//  //            clueContext.relationship match {
//  //              case "and" => mergeRddsWith(AND, prevRdd, currNodeRdd)(entity => entity.idNumber, (one, another) => one.mergeWithPlusingConditionCount(another))
//  //              case "or" => mergeRddsWith(OR, prevRdd, currNodeRdd)(entity => entity.idNumber, (one, another) => one.merge(another))
//  //              case "not" => mergeRddsWith(NOT, prevRdd, currNodeRdd)(entity => entity.idNumber, (one, another) => one.merge(another))
//  //            }
//  //
//  //          } else
//  //            spark.sparkContext.emptyRDD[ClueResultWithDetail]
//  //        }
//  //      }
//  //
//  //    //因为需求是阀值是对命中圈的限制,添加此mergeWithPlusingConditionCount,filter
//  //    val clueResultRdd: RDD[ClueResultWithDetail] =
//  //      if (relation.equalsIgnoreCase("and") && conditionsRange > 1)
//  //        clueOriginalResultRdd.filter(result => result.conditionCount >= threshold)
//  //      else
//  //        clueOriginalResultRdd
//  //    //保存到数据库当中
//  //    val finalResult: RDD[ClueResultWithDetail] = if (clueResultRdd.isEmpty()) spark.sparkContext.parallelize(List(bean.ClueResultWithDetail(clueContext.jobId, "0", "None", 0, null, 0, null, new Date(), null, null))) else clueResultRdd
//  //    val properties = new Properties()
//  //    properties.put("user", "root")
//  //    properties.put("password", "root")
//  //
//  //    val results = finalResult.collect()
//  //    //    for (item<-0 to results.length-1) {
//  //    //      println(results(item)+"##########")
//  //    //    }
//  //    import spark.implicits._
//  //    finalResult.toDF().write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.5.180:3306/ga", "t_clue_testnew", properties)
//  //
//  //    //    spark.createDataFrame(finalResult).write.mode(SaveMode.Append).jdbc("jdbc:mysql://10.68.2.237:3306/ga?characterEncoding=utf-8&serverTimezone=UTC", "TB_CLUE_RESULT", properties)
//  //    //    spark.createDataFrame(finalResult).write.mode(SaveMode.Append).jdbc("jdbc:mysql://10.68.2.237:3306/ga", "TB_CLUE_RESULT", properties)
//  //
//  //  }
//
//  private[service] def justClue(spark: SparkSession, jobId: String, conditions: Array[ClueConditionEntity], threshold: Int): RDD[ClueResultWithDetail] = {
//    require(conditions != null)
//    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
//    var rddWfifEntity: RDD[WifiEntity] = null
//    conditions.map(s => {
//      if (rddWfifEntity == null) {
//        val res = LoadESData.getOneDeviceData(spark, "lzwifi/wifi", s.deviceNumber, sdf.format(new Date(s.beginTime)), sdf.format(new Date(s.endTime)))
//        println(res.count() + "=================================")
//        if (res.count() > 0) {
//          rddWfifEntity = res.map(e => {
//
//            if (e._2.get("mac").get != None) //mac数据的封装
//              WifiEntity(
//
//                e._1,
//                e._2.get("mac").get + "",
//                e._2.get("mac").get + "", "MAC",
//                Timestamp.valueOf(e._2.get("captime").get + ""),
//                Timestamp.valueOf(e._2.get("captime").get + ""),
//                new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//                //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
//                e._2.get("device_code").get + "",
//                //                None,
//                //                "",
//                e._2.get("device_name").get + "",
//                e._2.get("address").get + "",
//                e._2.get("location").get + "",
//                e._2.get("captime").get + "",
//
//                java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//                java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//                e._2.get("phone").get + "",
//                "",
//                e._2.get("picpath").get + ""
//              )
//            else
//              WifiEntity(
//                e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "IMSI",
//                Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
//                //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
//                new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//                e._2.get("device_code").get + "",
//                //                None, ""
//                e._2.get("device_name").get + "",
//                e._2.get("address").get + "",
//                e._2.get("location").get + "",
//                e._2.get("captime").get + "",
//
//                java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//                java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//                e._2.get("phone").get + "",
//                "",
//                e._2.get("picpath").get + ""
//              )
//
//          })
//        } else {
//          println("未找到数据", "")
//        }
//      } else {
//        val res = LoadESData.getOneDeviceData(spark, "lzwifi/wifi", s.deviceNumber, sdf.format(new Date(s.beginTime)), sdf.format(new Date(s.endTime)))
//        if (res.count() > 0) {
//          rddWfifEntity = rddWfifEntity ++ res.map(e => {
//            if (e._2.get("mac").get != None) //mac数据的封装
//              WifiEntity(
//                e._1, e._2.get("mac").get + "", e._2.get("mac").get + "", "MAC",
//                Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
//                new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//                //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
//                e._2.get("device_code").get + "",
//                //                None, ""
//                e._2.get("device_name").get + "",
//                e._2.get("address").get + "",
//                e._2.get("location").get + "",
//                e._2.get("captime").get + "",
//
//                java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//                java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//                e._2.get("phone").get + "",
//                "",
//                e._2.get("picpath").get + ""
//              )
//            else
//              WifiEntity(
//                e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "IMSI",
//                Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
//                new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//                //                Timestamp.valueOf(e._2.get("ir_time").get + ""),
//                e._2.get("device_code").get + "",
//                e._2.get("device_name").get + "",
//                e._2.get("address").get + "",
//                e._2.get("location").get + "",
//                e._2.get("captime").get + "",
//
//                java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//                java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//                e._2.get("phone").get + "",
//                "",
//                e._2.get("picpath").get + ""
//                //                None, ""
//              )
//
//          })
//        } else {
//        }
//      }
//    })
//    //进行碰撞
//    val result: RDD[ClueResultWithDetail] = Try(rddWfifEntity.isEmpty()) match {
//      case Success(result) =>
//        if (!result)
//          new ClueRdd(rddWfifEntity).clueAsUnion(spark, jobId)
//        else spark.sparkContext.parallelize(List(bean.ClueResultWithDetail(jobId, "0", "None", 0, null, 0, "", new Date(), null, null,null))) //最新需求导致使用clueAsUnion,需求面试ClueEntity中的描述
//
//      case Failure(error) => spark.sparkContext.parallelize(List(bean.ClueResultWithDetail(jobId, "0", "None", 0, null, 0, null, new Date(), null, null,null)))
//    }
//    if (result != null && result.count() == 1 && result.collect().head.idNumber.equals("0"))
//      spark.sparkContext.emptyRDD[ClueResultWithDetail]
//    else
//      result
//  }
//
//
//  /**
//    * 将两个rdd根据定义好的relationship关系进行合并.合并通过对T提取key进行操作进行.
//    * 对于and操作,我们对one做链接another进行交集操作,并利用merge方法合并两个具有相同key的T为一个
//    * 对于or操作,我们对one与another进行全外链接操作,如果有交集的T,利用merge方法合并两个具有相同key的T为一个,否则添加元素
//    * 对于not操作,我们取another不在one中的内容
//    *
//    * @param relationShip
//    * @param one
//    * @param another
//    * @param key
//    * @param merge
//    * @tparam T
//    * @return
//    */
//  private[service] def mergeRddsWith[T: ClassTag : TypeTag](relationShip: Relationship, one: RDD[T], another: RDD[T])(key: T => String, merge: (T, T) => T): RDD[T] = {
//
//    val onePair =
//      one.mapPartitions(x => {
//        x.map(x => {
//          (key(x), x)
//        })
//      })
//    val anotherPair =
//      another.mapPartitions(x => {
//        x.map(x => {
//          (key(x), x)
//        })
//      })
//
//
//    relationShip match {
//      case AND => onePair.fullOuterJoin(anotherPair).map {
//        case (key, joinResult) => {
//          if (joinResult._1.isDefined && joinResult._2.isDefined)
//            merge(joinResult._1.get, joinResult._2.get)
//          else
//            joinResult._1.getOrElse(joinResult._2.get)
//        }
//      }
//      case OR => onePair.fullOuterJoin(anotherPair).map {
//        case (key, joinResult) => {
//          if (joinResult._1.isDefined && joinResult._2.isDefined)
//            merge(joinResult._1.get, joinResult._2.get)
//          else
//            joinResult._1.getOrElse(joinResult._2.get)
//        }
//      }
//      case NOT => onePair
//        .leftOuterJoin(anotherPair)
//        .filter(result => result._2._2.isEmpty)
//        .mapPartitions(x => {
//          x.map(_._2._1)
//        })
//      //        .optMap(_._2._1)
//    }
//  }
//
//  /**
//    * 对指定伴随分析条件进行分析,其中threshold表示某一个人在多个设备之间共同出现次数
//    *
//    * @param spark
//    * @param jobId
//    * @param conditions
//    * @param threshold
//    */
//  def following(spark: SparkSession, jobId: String, conditions: FollowingConditionEntity, threshold: Int = 2): Unit = {
//    //    import com.trs.media.spark.rdd.ExtensionRdd._
//    //    spark.conf.set("spark.sql.codegen.wholeStage",false)
//    require(conditions != null)
//    //将条件转换成hybase的检索表达
//
//    val beginTimeStr = DateFormat.dateToString(new Date(conditions.beginTime))
//    val endTimeStr = DateFormat.dateToString(new Date(conditions.endTime))
//    var list: util.ArrayList[String] = new util.ArrayList()
//    conditions.deviceNumbers.split(",").map(s => list.add(s))
//    //    conditions.deviceNumbers.split(",").toList
//    var flowRDD: RDD[WifiEntity] = null;
//
//    val esrdd: RDD[(String, Map[String, AnyRef])] = LoadESData.getMutliDeviceData(spark, "lzwifi/wifi", list, beginTimeStr, endTimeStr)
//    if (esrdd.count() > 0) {
//      flowRDD = esrdd.map(e => {
//        if (e._2.get("mac").get != None) //mac数据的封装
//          WifiEntity(
//            e._1, e._2.get("mac").get + "",
//            e._2.get("mac").get + "", "MAC",
//            Timestamp.valueOf(e._2.get("captime").get + ""),
//            Timestamp.valueOf(e._2.get("captime").get + ""),
//            new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//            //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
//            e._2.get("device_code").get + "",
//            e._2.get("device_name").get + "",
//            e._2.get("address").get + "",
//            e._2.get("location").get + "",
//            e._2.get("captime").get + "",
//
//            java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//            java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//            e._2.get("phone").get + "",
//            "",
//            e._2.get("picpath").get + ""
//            //            None, ""
//
//          )
//        else
//          WifiEntity(
//            e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "IMSI",
//            Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
//            new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
//            //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
//            e._2.get("device_code").get + "",
//            e._2.get("device_name").get + "",
//            e._2.get("address").get + "",
//            e._2.get("location").get + "",
//            e._2.get("captime").get + "",
//            java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
//            java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
//
//
//            e._2.get("phone").get + "",
//            "",
//            e._2.get("picpath").get + ""
//            //            None, ""
//          )
//      });
//    } else {
//      println("未找到相应的数据")
//    }
//    val rdd: RDD[WifiEntityWithLongInsteadOfTimestamp] = flowRDD
//      .mapPartitions(x => {
//        x.map(wifiEntity =>
//          WifiEntityWithLongInsteadOfTimestamp(wifiEntity.id,
//            wifiEntity.showedIdNumber,
//            wifiEntity.idNumber,
//            wifiEntity.idNumberType,
//            wifiEntity.behaviorStartTime.getTime,
//            wifiEntity.behaviorEndTime.getTime,
//            wifiEntity.deviceNumber,
//            wifiEntity.address,
//            wifiEntity.location,
//            wifiEntity.deviceName,
//            wifiEntity.captime,
//            wifiEntity.phoneNumber,
//            wifiEntity.latitude,
//            wifiEntity.longitude,
//            wifiEntity.picpath
//          ))
//      })
//    //      .map(wifiEntity =>
//    //        WifiEntityWithLongInsteadOfTimestamp(wifiEntity.id, wifiEntity.showedIdNumber,
//    //          wifiEntity.idNumber, wifiEntity.idNumberType,
//    //          wifiEntity.behaviorStartTime.getTime, wifiEntity.behaviorEndTime.getTime,
//    //          wifiEntity.deviceNumber))
//    rdd.persist()
//    //进行伴随分析
//    val result: Dataset[FollowingResult] = Try(rdd.count()) match {
//      case Success(count) if count > 0 => doFollowingAnalysisAsRDD(spark, jobId, conditions, rdd, threshold)
//      case Failure(error) => { //如果加载hybase数据失败,则返回无碰撞结果的信息
//        import spark.implicits._
//        spark.sparkContext.parallelize(List(FollowingResult(jobId, "0", "None", 0, error.getMessage))).toDS()
//      }
//
//    }
//    val count = result.count()
//    val results = result.collect()
//    for (i <- 0 to results.length - 1) {
//      println(results(i).conditions + results(i).count + results(i).idNumber + "....")
//    }
//    if (count > 0) {
//      //      log.infoLog(s"伴随分析结果", s"分析结果数量为:${result.count()}")
//      //将结果保存到mysql当中
//      val properties = new Properties()
//      properties.put("user", "root")
//      properties.put("password", "root")
//
//
//      result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.5.180:3306/ga", "t_follow_testnew", properties)
//    }
//  }
//
//  def doFollowingAnalysisAsRDD(spark: SparkSession, jobId: String, conditions: FollowingConditionEntity, wifiEntities: RDD[WifiEntityWithLongInsteadOfTimestamp], threshold: Int = 2): Dataset[FollowingResult] = {
//    val filter = wifiEntities
//      .filter(entity => entity.idNumber.equals(conditions.idNumber))
//    println(filter.count() + "filter")
//    val followingConditions = filter
//      .mapPartitions(x => {
//        x.map(entity =>
//          (entity.deviceNumber, entity.behaviorStartTime - conditions.slice * 1000, entity.behaviorStartTime + conditions.slice * 1000)
//        )
//      })
//      //      .map(entity =>
//      //        (entity.deviceNumber, entity.behaviorStartTime - conditions.slice * 1000, entity.behaviorStartTime + conditions.slice * 1000))
//      .collect()
//    val finalResultRdd =
//      if (followingConditions.isEmpty) {
//        spark.sparkContext.parallelize(List(FollowingResult(jobId, "0", "None", 0, "")))
//      }
//      else {
//        //        log.infoLog("!!!!!!!!!!!!!!!!!!!!!!!!", s"${conditions.idNumber}----->存在${followingConditions.size}条轨迹,其中一条为:${followingConditions.head.toString()}")
//        val followingBehaviorConditions = spark.sparkContext.broadcast(followingConditions)
//        val value: RDD[(String, WifiEntityWithLongInsteadOfTimestamp)] = wifiEntities.filter(entity => { //得到所有符合伴随条件的人
//          val broadcastConditions = followingBehaviorConditions.value
//          broadcastConditions.filter { case (device, start, end) => {
//            (!entity.idNumber.equals(conditions.idNumber)) &&
//              entity.deviceNumber.equals(device) && entity.behaviorStartTime >= start && entity.behaviorStartTime <= end
//          }
//          }.size > 0
//        })
//
//          .map(entity => (entity.idNumber, entity))
//
//
//        val resultRdd = value
//          .aggregateByKey(List.empty[WifiEntityWithLongInsteadOfTimestamp])(
//            (items, item) => items :+ item,
//            (listOne, listAnother) => {
//              listOne.foldLeft(listAnother)((items, item) => {
//                items :+ item
//              })
//            })
//
//          .filter(_._2.size >= threshold)
//          .map {
//            case (idNumber, entities) => {
//              val detailsGroupByDeviceNumber = entities.groupBy(entity => entity.deviceNumber) //对所有关系人所出现的设备号进行排序,如果出现在的设备地址不足阀值,则舍弃
//              if (detailsGroupByDeviceNumber.keySet.size >= threshold)
//                Some(FollowingResult(jobId, idNumber, entities.head.idNumberType, entities.size, entities.map(_.id).mkString(",")))
//              else
//                None
//            }
//          }
//          .filter(result => result.isDefined)
//          .map(_.get)
//        //        log.infoLog("!!!!!!!!!!!!!!!!!!!!", s"存在${resultRdd.count()}个人存在伴随关系")
//        if (resultRdd.isEmpty())
//          spark.sparkContext.parallelize(List(FollowingResult(jobId, "0", "None", 0, "")))
//        else
//          resultRdd
//      }
//    import spark.implicits._
//    finalResultRdd.toDS()
//  }
//
//  /**
//    * 根据传入的海贝检索表达式获取加载数据
//    *
//    * @param spark
//    * @param where
//    * @return
//    */
//
//  /**
//    * 将传入的ClueConditionEntity转换成海贝的检索表达式,多个条件之间为或的关系
//    *
//    * @param conditions
//    * @return
//    */
//  private[this] def buildTrssl(conditions: Array[ClueConditionEntity]): String = {
//    conditions.map(condition => {
//      val beginTime = DateFormat.dateToString(new Date(condition.beginTime))
//      val endTime = DateFormat.dateToString(new Date(condition.endTime))
//      if (condition.didAppear)
//        s"""(device_code:${condition.deviceNumber} AND captime:["${beginTime}" TO "${endTime}"] AND repeat:0)"""
//      else
//        s"""(device_code:${condition.deviceNumber} AND captime:["${beginTime}" TO "${endTime}"] AND repeat:0)"""
//    }).mkString(" OR ")
//  }
//
//
//  /**
//    * 将传入的ClueConditionEntity转换成Dataframe的检索表达式,多个条件之间为或的关系
//    *
//    * @param conditions
//    * @return
//    */
//  private[this] def buildDataFrameWhere(conditions: Array[ClueConditionEntity]): String = {
//    conditions.map(condition => {
//      if (condition.didAppear)
//        s"""(TABLE=='system.wifi' AND DEVICE_ID=='${condition.deviceNumber}' AND BSTART_TIME_MS>=${condition.beginTime} AND BSTART_TIME_MS <= ${condition.endTime})"""
//      else
//        s"""(TABLE=='system.wifi' AND DEVICE_ID=='${condition.deviceNumber}' AND BSTART_TIME_MS>=${condition.beginTime} AND BSTART_TIME_MS <= ${condition.endTime})"""
//    }).mkString(" OR ")
//  }
//}
//
//case class HdfsConditionEntity(where: String, beginTime: Long, endTime: Long, didAppear: Boolean = true)
//
///**
//  * 碰撞条件的封装
//  *
//  * @param deviceNumber 设备编号
//  * @param beginTime    开始时间
//  * @param endTime      结束时间
//  * @param didAppear    是否出现在该设备内
//  */
//case class ClueConditionEntity(deviceNumber: String, beginTime: Long, endTime: Long, didAppear: Boolean)
//
///**
//  * 伴随分析所需条件
//  *
//  * @param idType        证件类型,常量为:MAC/IMSI/IMEI
//  * @param idNumber      证件号码
//  * @param beginTime     分析开始时间
//  * @param endTime       分析结束时间
//  * @param slice         时间滑动时间
//  * @param deviceNumbers 伴随需要分析的设备编号
//  */
//case class FollowingConditionEntity(idType: String, idNumber: String, beginTime: Long, endTime: Long, slice: Int, deviceNumbers: String, hasliferow: String, fuzzyway: String, activityTimes: List[ActivityTime])
//
///**
//  * 将wifientity的时间条件转换成long型,方便进行时间的查询工作
//  *
//  * @param id
//  * @param showedIdNumber
//  * @param idNumber
//  * @param idNumberType
//  * @param behaviorStartTime
//  * @param behaviorEndTime
//  * @param deviceNumber
//  */
//case class WifiEntityWithLongInsteadOfTimestamp(id: String,
//                                                showedIdNumber: String,
//                                                idNumber: String,
//                                                idNumberType: String,
//                                                behaviorStartTime: Long,
//                                                behaviorEndTime: Long,
//                                                deviceNumber: String,
//                                                address: String,
//                                                location:String,
//                                                devicename: String,
//                                                captime: String,
//                                                phone:String,
//                                                latitude: java.lang.Double,
//                                                longitude: java.lang.Double,
//                                                picpath:String
//                                               )
//
///**
//  * 伴随结果封装
//  *
//  * @param jobId
//  * @param idNumber
//  * @param idNumberType
//  * @param count      伴随次数
//  * @param conditions 伴随分析条件,由被分析账号的轨迹信息生成
//  */
//case class FollowingResult(jobId: String, idNumber: String, idNumberType: String, count: Long, conditions: String)
//
//case class FollowResult(
//                         resultid: String,
//                         codetype: String,
//                         //                         conditions: String,
//                         count: Integer,
//                         createtime: String,
//                         jobid: Integer,
//                         tagcode: String,
//                         address: String,
//                         location:String,
//                         phone: String
//                       )
//
//case class FollowResultWithDetail(
//                                   resultid: String,
//                                   codetype: String,
//                                   //                         conditions:String,
//                                   count: Integer,
//                                   createtime: String,
//                                   jobid: Integer,
//                                   tagcode: String,
//                                   address: String,
//                                   location:String,
//                                   phone: String,
//                                   detail: util.ArrayList[FollowDetail] = new util.ArrayList[FollowDetail]()
//                                 )
//
//case class FollowDetail(
//                         //                       id:Integer,
//                         captime: String,
//                         codetype: String,
//                         devicecode: Integer,
//                         devicename: String,
//                         address: String,
//                         phone: String,
//                         var resultid: String,
//                         tagcode: String,
//                         latitude: java.lang.Double,
//                         //                         longitude: java.lang.Double
//                         longtitude: java.lang.Double,
//                         picpath: String
//                       ) {
//  def setResultId(string: String) = {
//    resultid = string
//  }
//}